package com.navinfo.tripanalysis.offline.service.impl;

import com.google.common.collect.Lists;
import com.navinfo.tripanalysis.common.pojo.Point0200;
import com.navinfo.tripanalysis.common.pojo.Point0F37;
import com.navinfo.tripanalysis.common.util.DateUtils;
import com.navinfo.tripanalysis.offline.pojo.AlgorithmOuterData;
import com.navinfo.tripanalysis.offline.service.PointProtocolConvertService;
import com.navinfo.tripanalysis.offline.service.SavePointProtocolService;
import lombok.Data;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * 通过0200和0F37计算行程统计版本时，使用的公共位置协议数据落盘服务<br/>
 * 保存公共位置协议服务的实现<br/>
 * @author web
 */
@Data
public class SavePointProtocolServiceImpl extends AbstractHiveSaveService implements SavePointProtocolService {

    /**
     * 适配统一协议落盘数据结构服务
     */
    private PointProtocolConvertService service;
    @Override
    public void save(SparkSession spark, JavaPairRDD<Long, AlgorithmOuterData> pairTidOutData, long day) {
        long start = System.currentTimeMillis();
        logger.error("进行公共轨迹协议落盘，保存Hive开始...");

        String dayStr = DateUtils.format(new Date(day), DateUtils.DateFormat.YYYYMMDD);
        super.saveData(spark, pairTidOutData, null, true, service.getPartitionCols(),  Lists.newArrayList("'"+dayStr+"'","tid%32"));

        logger.error("进行公共轨迹协议落盘，保存Hive结束，耗{}ms", System.currentTimeMillis()-start);
    }

    @Override
    protected StructType createSchema() {
        return service.createSchema(service.createStructTypeList());
    }

    @Override
    protected JavaRDD<Row> createRowRDD(JavaPairRDD resourceRDD) {
        return resourceRDD.flatMap((FlatMapFunction<Tuple2<Long, AlgorithmOuterData>, Row>) tuple2 -> {
            AlgorithmOuterData outData = tuple2._2;
            List<Row> row0200 = Optional.ofNullable(outData.getPoints0200()).orElse(new ArrayList<>())
                    .stream().filter(e -> e.getTripId()!=0).map(e -> service.toRow(com.navinfo.tripanalysis.common.util.PointUtils.toPointProtocol((Point0200)e))).collect(Collectors.toList());
            List<Row> row0f37 = Optional.ofNullable(outData.getPoints0f37()).orElse(new ArrayList<>())
                    .stream().filter(e -> e.getTripId()!=0).map(e -> service.toRow(com.navinfo.tripanalysis.common.util.PointUtils.toPointProtocol((Point0F37)e))).collect(Collectors.toList());
            row0200.addAll(row0f37);
            return row0200.iterator();
        });

    }
}
